package com.example.emos.wx.task;

import com.example.emos.wx.db.pojo.MessageEntity;
import com.example.emos.wx.db.pojo.MessageRefEntity;
import com.example.emos.wx.exception.EmosException;
import com.example.emos.wx.service.MessageService;
import com.rabbitmq.client.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;

/**
 * @ClassName MessageTask
 * @Description TODO
 * @Author daoketa
 * @Version 1.0
 * 收发RabbitMQ消息
 **/
@Slf4j
@Component
public class MessageTask {
    @Autowired
    private ConnectionFactory factory;
    @Autowired
    private MessageService messageService;

    /**
     * 同步发送消息
     * @author daoketa
     * @param
     */
    public void send(String topic, MessageEntity messageEntity){
        /** 向MongoDB保存消息数据，返回消息ID */
        String id = messageService.insertMessage(messageEntity);
        /** 向RabbitMQ发送消息 */
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel();){

            /** 连接到某个Topic */
            channel.queueDeclare(topic, true, false, false, null);

            /** 创建AMQP协议参数对象，添加附加属性 */
            HashMap header = new HashMap(); //存放属性数据
            header.put("messageId", id);
            AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().headers(header).build();

            channel.basicPublish("", topic, properties, messageEntity.getMsg().getBytes());
            log.debug("消息发送成功");
        }catch (Exception e){
            log.error("执行异常", e);
            throw new EmosException("向MQ发送消息失败");
        }
    }

    /**
     * 异步发送消息
     * @author daoketa
     * @date 2023/3/3 20:08
     * @param
     */
    @Async
    public void sendAsync(String topic, MessageEntity messageEntity) {
        send(topic, messageEntity);
    }

    /**
     * 同步接收消息
     * @author daoketa
     * @param
     */
    public Integer receive(String topic){
        Integer count = 0;
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel();){

            /** 从队列中获取消息，不自动确认 */
            channel.queueDeclare(topic, true, false, false, null);

            /** Topic中有多少条数据未知，所以使用死循环接收数据，直到接收不到消息，退出死循环 */
            while(true){
                /** 创建响应接收数据，禁止自动发送Ack应答 */
                GetResponse response = channel.basicGet(topic, false);
                if (response != null){
                    /** 获取附加属性对象 */
                    AMQP.BasicProperties properties = response.getProps();
                    Map<String, Object> headers = properties.getHeaders();
                    String messageId = headers.get("messageId").toString();

                    /** 获取消息正文 */
                    byte[] body = response.getBody();
                    String message = new String(body);
                    log.debug("从RabbitMQ接收的消息：" + message);

                    /** 把消息存储在MongoDB中 */
                    MessageRefEntity entity = new MessageRefEntity();
                    entity.setMessageId(messageId);
                    entity.setReceiverId(Integer.parseInt(topic));
                    entity.setReadFlag(false);
                    entity.setLastFlag(true);
                    messageService.insertRef(entity);

                    /** 数据保存到MongoDB后，才发送Ack应答，让Topic删除这条消息 */
                    long deliveryTag = response.getEnvelope().getDeliveryTag();
                    channel.basicAck(deliveryTag, false);

                    count++;
                }else {
                    break; //接收不到消息，则退出死循环
                }
            }
        }catch (Exception e){
            log.error("执行异常", e);
        }
        return count;
    }

    /**
     * 异步接收消息
     * @author daoketa
     * @date 2023/3/3 20:21
     * @param
     */
    @Async
    public int receiveAysnc(String topic) {
        return receive(topic);
    }

    /**
     * 同步删除队列
     * @author daoketa
     * @date 2023/3/3 20:22
     * @param
     */
    public void deleteQueue(String topic) {
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            channel.queueDelete(topic);
            log.debug("消息队列成功删除");
        } catch (Exception e) {
            log.error("删除队列失败", e);
            throw new EmosException("删除队列失败");
        }
    }

    /**
     * 异步删除队列
     * @author daoketa
     * @date 2023/3/3 20:22
     * @param
     */
    @Async
    public void deleteQueueAsync(String topic) {
        deleteQueue(topic);
    }

}
